package xin.marcher.module.inventory.manage;

import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.util.CollectionUtils;
import xin.marcher.module.common.constants.inventory.TairInventoryConstant;
import xin.marcher.module.common.tair.TairCache;
import xin.marcher.module.common.utils.DateFormatUtil;
import xin.marcher.module.common.utils.JsonUtil;
import xin.marcher.module.inventory.domain.bo.BucketCacheBO;
import xin.marcher.module.inventory.domain.bo.BucketLocalCache;
import xin.marcher.module.inventory.domain.bo.CalculateInventory;

import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.stream.Collectors;

/**
 * 多线程消费队列 计算分桶下线阈值
 */
@Slf4j
public class CalculateThresholdRunner implements Runnable {
    /**
     * 处理内部消息的队列
     */
    private BlockingQueue blockingQueue;

    private TairCache tairCache;

    public CalculateThresholdRunner(BlockingQueue blockingQueue, TairCache tairCache) {
        this.blockingQueue = blockingQueue;
        this.tairCache = tairCache;
    }

    /**
     * 内部线程计算每个商品的缓存信息
     */
    @Override
    public void run() {
        try {
            while (true) {
                BucketLocalCache bucketLocalCache = (BucketLocalCache) blockingQueue.take();
                String currentDate = DateFormatUtil.formatDateTime();
                // 获取可用列表，不可用列表默认触发下线，库存暂不考虑计算入内，否则增加性能开销
                List<BucketCacheBO> availableList = bucketLocalCache.getAvailableList();
                if (CollectionUtils.isEmpty(availableList)) {
                    return;
                }
                // 获取可用列表的分桶缓存集合key
                List<String> cacheKeyList = availableList.stream().map(BucketCacheBO::getBucketNo).collect(Collectors.toList());
                // 返回批量的缓存数量
                List<String> bucketNumList = tairCache.mget(cacheKeyList);
                // 获取中心桶库存的key
                String sellerInventoryKey = buildSellerInventoryKey(bucketLocalCache.getSellerId(), bucketLocalCache.getSkuId());
                // 剩余库存
                Integer residueNum = 0;
                String inventoryNum = tairCache.get(sellerInventoryKey);
                if (!StringUtils.isEmpty(inventoryNum)) {
                    residueNum = residueNum + Integer.parseInt(inventoryNum);
                }
                // 合并累计相加，得到当前商品对应的总库存
                for (String bucketNum : bucketNumList) {
                    if (!StringUtils.isEmpty(bucketNum)) {
                        residueNum = residueNum + Integer.parseInt(bucketNum);
                    }
                }
                //获取之前存储的库存信息，计算差集得到实际库存消费速度
                String key = buildSellerInventoryResidueKey(bucketLocalCache.getSellerId(), bucketLocalCache.getSkuId());
                // 上次计算存储的库存实际值以及查询时间，用于计算均匀的每秒库存消费
                String oldCalculateInventory = tairCache.get(key);
                // 默认为当前配置的分桶下线阈值
                Integer thresholdValue = bucketLocalCache.getInventoryBucketConfig().getThresholdValue();
                if (!StringUtils.isEmpty(oldCalculateInventory)) {
                    CalculateInventory calculateInventory = JsonUtil.json2Object(oldCalculateInventory, CalculateInventory.class);
                    //计算上一次的库存 减去此次的库存，得到这个时间点消耗了多少库存
                    int diffNum = calculateInventory.getOldResidueNum() - residueNum;
                    // 上一次计算的创建时间
                    String createDate = calculateInventory.getCreateDate();
                    // 当前时间 减去上一次的计算的创建时间，得到间隔时间差，再通过差集的库存除以秒，得到每秒平均的消耗库存
                    long consumeStock = diffNum / (Long.parseLong(currentDate) - Long.parseLong(createDate));
                    if (consumeStock > 0) {
                        //  每秒消耗的库存 / 当前存活的分桶数量，得到目前分桶的下线阈值
                        Long newThresholdValue = consumeStock / availableList.size();

                        // 这里计算的下线阈值，最小值不能低于配置的最小阈值
                        if (newThresholdValue > thresholdValue) {
                            thresholdValue = newThresholdValue.intValue();
                            // 阈值的最大值，不能超过库存深度的10%比例
                            int maxDepthNum = bucketLocalCache.getInventoryBucketConfig().getMaxDepthNum() / bucketLocalCache.getInventoryBucketConfig().getThresholdProportion();
                            if (thresholdValue > maxDepthNum) {
                                thresholdValue = maxDepthNum;
                            }
                        }
//                        log.error("预估的分桶下线阈值{},实际使用的分桶下线阈值{}",newThresholdValue,thresholdValue);
                    }
                }

                String thresholdKey = buildSellerInventoryThresholdKey(bucketLocalCache.getSellerId(), bucketLocalCache.getSkuId());

                // 存储该商品的预估下线阈值
                tairCache.set(thresholdKey, String.valueOf(thresholdValue), 0);
                // 存储一下这次计算的库存和时间
                CalculateInventory calculateInventory = new CalculateInventory();
                calculateInventory.setOldResidueNum(residueNum);
                calculateInventory.setCreateDate(DateFormatUtil.formatDateTime());
                tairCache.set(key, JSONObject.toJSONString(calculateInventory), 0);
            }
        } catch (Exception e) {
            log.error("处理缓存下线阈值异常", e);
        }
    }

    /**
     * 中心桶库存的key
     *
     * @param sellerId
     * @param skuId
     * @return
     */
    private String buildSellerInventoryKey(String sellerId, String skuId) {
        return TairInventoryConstant.SELLER_INVENTORY_PREFIX + sellerId + skuId;
    }


    /**
     * 卖家的商品剩余库存
     *
     * @param sellerId
     * @param skuId
     * @return
     */
    private String buildSellerInventoryResidueKey(String sellerId, String skuId) {
        return TairInventoryConstant.SELLER_INVENTORY_RESIDUE_PREFIX + sellerId + skuId;
    }


    /**
     * 卖家的商品分桶下线最大消耗时间
     *
     * @param sellerId
     * @param skuId
     * @return
     */
    private String buildSellerInventoryOfflineKey(String sellerId, String skuId) {
        return TairInventoryConstant.SELLER_BUCKET_OFFLINE_TIME_PREFIX + sellerId + skuId;
    }

    /**
     * 卖家的商品 分桶下线阈值
     *
     * @param sellerId
     * @param skuId
     * @return
     */
    private String buildSellerInventoryThresholdKey(String sellerId, String skuId) {
        return TairInventoryConstant.SELLER_INVENTORY_THRESHOLD_PREFIX + sellerId + skuId;
    }
}
